#include "channel_task.h"
#include "downlink_connect.h"
#include <future>
#include <settings/config_parser.h>
#include <dbms/dbms_acq_data.h>
namespace data_acq {
ChannelTask::ChannelTask(const dbms::DeviceChannel_t &channel)
	: deviceChannel_(channel)
{
	if (deviceChannel_.ctype() == dbms::CT_COM) {
		::dbms::Serial_t *com = deviceChannel_.mutable_com();
		settings::ConfigParser configParser;
		settings::ConfigParser::Channel chnn =
			configParser.FindChannel(com->name());
		HTELINK_LOG_INFO("channel %d: %s -> %s", deviceChannel_.id(),
			com->name(), chnn.devPath);
		com->set_name(chnn.devPath);
	}
	downlink_ = downlink::DownlinkConnect::ConnectFactory(deviceChannel_);
	uvLoop_ = uv_loop_new();
}

ChannelTask::~ChannelTask()
{
	Stop();
	uv_loop_delete(uvLoop_);
	uvLoop_ = nullptr;
}
int32_t ChannelTask::ID() { return deviceChannel_.id(); }

bool ChannelTask::HasDevice(const std::string &deviceId)
{
	auto it = std::find_if(deviceTaskList_.begin(), deviceTaskList_.end(),
		[&](const std::shared_ptr<DeviceTask> &task) {
			return deviceId == task->ID();
		});
	return (it != deviceTaskList_.end());
}

void ChannelTask::Subscribe(SubscribeCallback subscrib)
{
	subsribeFunc_ = subscrib;
}

void ChannelTask::SubscribeException(SubscribeExceptionCallback subscribe)
{
	subsribeExceptionFunc_ = subscribe;
}

void ChannelTask::Start(bool detach)
{

	Stop();
	HTELINK_TRACE("Channel %d Start", deviceChannel_.id());
	OnGetDevices(deviceTaskList_);
	run_.store(true, std::memory_order_release);
	uvTimer_.data = this;
	uv_timer_init(uvLoop_, &uvTimer_);
	uvTimerUp_.data = this;
	uv_timer_init(uvLoop_, &uvTimerUp_);

	threadTask_ = std::thread([&]() { Run(); });
	if (detach) {
		threadTask_.detach();
	} else {
		threadTask_.join();
	}
}

void ChannelTask::Stop()
{
	HTELINK_TRACE("Channel %d Stop", deviceChannel_.id());
	run_.store(false, std::memory_order_release);

	if (IsRunning()) {
		uv_async_send(&uvAsync_);
		uv_timer_stop(&uvTimer_);
		uv_timer_stop(&uvTimerUp_);
		uv_walk(
			uvLoop_,
			[](uv_handle_t *handle, void *arg) {
				if (!uv_is_closing(handle)) {
					HTELINK_LOG_DEBUG(
						"uv close %d", (int32_t)uv_handle_get_type(handle));
					uv_close(handle, nullptr);
				}
			},
			nullptr);
		uv_run(uvLoop_, UV_RUN_NOWAIT);
	}

	if (threadTask_.joinable()) {
		threadTask_.join();
	}
	deviceTaskList_.clear();
}

bool ChannelTask::IsRunning() const
{
	if (uvLoop_ == nullptr) {
		return false;
	}
	return uv_loop_alive(uvLoop_);
}

void ChannelTask::Run()
{
	HTELINK_TRACE("Channel %d Run", deviceChannel_.id());
	uv_async_init(
		uvLoop_, &uvAsync_, [](uv_async_t *handle) { uv_stop(handle->loop); });
	uvImmed_.data = this;
	uv_async_init(uvLoop_, &uvImmed_, [](uv_async_t *handle) {
		ChannelTask *task = static_cast<ChannelTask *>(handle->data);
		ASSERT(task != nullptr);
		task->DoAcq();
	});

	uv_timer_cb timerCb = [](uv_timer_t *handle) {
		ChannelTask *task = static_cast<ChannelTask *>(handle->data);
		ASSERT(task != nullptr);
		task->OnPeriodTimeout();
	};
	uv_timer_start(&uvTimer_, timerCb, 5000, deviceChannel_.reportperiod());
	uv_timer_cb timerCb2 = [](uv_timer_t *handle) {
		ChannelTask *task = static_cast<ChannelTask *>(handle->data);
		ASSERT(task != nullptr);
		task->OnPeriodUpdate();
	};
	uv_timer_start(&uvTimerUp_, timerCb2, 60000, 5000);
	uv_run(uvLoop_, UV_RUN_DEFAULT);
}

void ChannelTask::DoAcq()
{
	while (run_.load(std::memory_order_acquire)) {
		AcqInfo acq;
		if (!PopQueue(acq)) {
			break;
		}
		std::shared_ptr<DeviceTask> deviceTask = acq.device.lock();
		if (deviceTask == nullptr) {
			HTELINK_LOG_ERR("device has been free");
			continue;
		}

		deviceTask->OnAcq({acq.subsribeFunc, acq.subsribeExceptionFunc,
							  [&](const libxcom::CmdValue &recv,
								  const ParseRecv &parseRecv) -> bool {
								  return OnDownlink(recv, parseRecv);
							  }},
			acq.funcPoints, acq.params);
	}
}

bool ChannelTask::PopQueue(AcqInfo &acq)
{
	std::lock_guard<std::mutex> queueLock(queueMutex_);
	if (deviceTaskQueue_.empty()) {
		return false;
	}
	acq = deviceTaskQueue_.front();
	deviceTaskQueue_.pop();
	return true;
}

void ChannelTask::PushQueue(const AcqInfo &acqInfo)
{
	std::lock_guard<std::mutex> queueLock(queueMutex_);
	deviceTaskQueue_.push(acqInfo);
}

void ChannelTask::CommitImmedAcq(const std::string &deviceId,
	const std::vector<dbms::AcqItem_t> &items, SubscribeCallback subsribeFunc,
	SubscribeExceptionCallback subsribeExceptionFunc)
{
	auto it = std::find_if(deviceTaskList_.begin(), deviceTaskList_.end(),
		[&](const std::shared_ptr<DeviceTask> &task) {
			return deviceId == task->ID();
		});
	if (it == deviceTaskList_.end()) {
		HTELINK_LOG_ERR("not find device %s", deviceId);
		return;
	}
	AcqInfo acqInfo;
	acqInfo.device = *it;
	acqInfo.subsribeFunc = subsribeFunc;
	acqInfo.subsribeExceptionFunc = subsribeExceptionFunc;
	for (auto &item : items) {
		dbms::Funcpoint_t funcpoint = (*it)->GetFuncpoint(item.funcid());
		if (!funcpoint.id().empty()) {
			acqInfo.funcPoints.push_back(funcpoint);
		}
	}
	if (acqInfo.funcPoints.empty()) {
		acqInfo.funcPoints = (*it)->GetAcqFuncpoints();
	}
	PushQueue(acqInfo);
	uv_async_send(&uvImmed_);
}

std::vector<uint8_t> ChannelTask::DownlinkWork(const std::vector<uint8_t> &cmd)
{
	std::vector<uint8_t> result;
	OnDownlink(cmd, [&](uint8_t *buf, int32_t len) -> libxcom::RECV_RESULT {
		result.assign(buf, buf + len);
		return RECV_COMPLETE;
	});
	return result;
}

void ChannelTask::OnPeriodTimeout()
{
	HTELINK_TRACE("Period Acq %d, device count: %d", deviceChannel_.id(),
		deviceTaskList_.size());
	OnGetPeriodAcq();
	DoAcq();
}

void ChannelTask::OnGetPeriodAcq()
{
	for (auto &device : deviceTaskList_) {
		AcqInfo acqInfo;
		acqInfo.device = device;
		acqInfo.subsribeFunc = [&](const dbms::Device_t &device,
								   const dbms::AcqData_t &acq) -> bool {
			dbms::DbmsAcqdata dbmsAcq(VENDOR_DB_PATH + "/" + device.id());
			dbmsAcq.Save(acq);
			return true;
		};
		acqInfo.subsribeExceptionFunc = subsribeExceptionFunc_;
		acqInfo.funcPoints = device->GetAcqFuncpoints();
		if (!acqInfo.funcPoints.empty()) {
			PushQueue(std::move(acqInfo));
		}
	}
}

bool ChannelTask::OnDownlink(
	const libxcom::CmdValue &recv, const ParseRecv &parseRecv)
{
	HTELINK_TRACE("On Downlink");
	if (downlink_ == nullptr) {
		HTELINK_LOG_ERR("downlink has been free");
		return false;
	}

	std::atomic<bool> setFlag{false};
	std::promise<bool> isFinish;
	auto future = isFinish.get_future();
	downlink_->OnReceive([&, parseRecv](uint8_t recv[], int len) {
		auto result = parseRecv(recv, len);
		if (result == RECV_FAILED || result == RECV_COMPLETE) {
			if (!setFlag.exchange(true)) {
				isFinish.set_value(result == RECV_COMPLETE);
			}
		}
	});
	downlink_->OnError([&](int32_t err) {
		if (!setFlag.exchange(true)) {
			isFinish.set_value(err == 0);
		}
	});
	bool result = downlink_->Send(uvLoop_, recv.Cmd(), recv.Length());
	if (!result) {
		HTELINK_LOG_ERR("Send failed");
		return false;
	}
	future.wait_for(std::chrono::milliseconds(deviceChannel_.timeout()));
	downlink_->Stop();
	return true;
}

void ChannelTask::OnGetDevices(
	std::vector<std::shared_ptr<DeviceTask> > &deviceTaskList)
{
	dbms::DbmsDevice dbmsDevice;
	dbmsDevice.Get([&](const dbms::Device_t &device) {
		if (deviceChannel_.id() == device.channelid()) {
			deviceTaskList.emplace_back(std::make_shared<DeviceTask>(device));
		}
	});
}

void ChannelTask::OnPeriodUpdate()
{
	for (auto &device : deviceTaskList_) {
		dbms::DbmsAcqdata dbmsAcq(VENDOR_DB_PATH + "/" + device->ID());
		dbms::AcqData_t acq;
		if (!dbmsAcq.GetOldest(acq)) {
			continue;
		}
		if (subsribeFunc_ == nullptr) {
			break;
		}
		if (subsribeFunc_(device->GetDevice(), acq)) {
			dbmsAcq.Drop(acq);
		}
	}
}

} // namespace data_acq